/**
    自定义实现的一个发送器，用来向kafka发送数据：
 */
package com.demo.kafka.flink.shop;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

public class FlinkSource implements SourceFunction<String> {

    private boolean isRunning = true;

    /**
     * 在run方法中实现一个循环，模拟产生购物数据
*/
    @Override
    public void run(SourceFunction.SourceContext<String> ctx) throws Exception {
        while(isRunning){

            //模拟产生购买手机的数据
            //创建一个数组队列phone，在里面添加5类手机品牌
            List<String> phone = new ArrayList<>();
            phone.add("Samsung");
            phone.add("华为");
            phone.add("iPhone");
            phone.add("小米");
            phone.add("OPPO");

            //产生0-4随机索引值，通过该索引值来访问phone中的元素
            int i = new Random().nextInt(5);
            ctx.collect(phone.get(i));

            //每2毫秒产生一条数据
            Thread.sleep(200);
        }
    }

    //取消一个cancel的时候会调用的方法
    @Override
    public void cancel() {
        isRunning = false;
    }
}